package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradeOrderBean;
import com.bw.gmall.realtime.common.bean.TradeProvinceOrderBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import com.bw.gmall.realtime.common.util.HBaseUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.time.Duration;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

public class DwsTradeProvinceOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeProvinceOrderWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,Constant.DWS_TRADE_PROVINCE_ORDER_WINDOW,4,10030);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {


        //1）从Kafka下单明细主题读取数据
//        dataStreamSource.print();


        //2）过滤null数据并转换数据结构 etl
        SingleOutputStreamOperator<JSONObject> eltedStream = eltStream(dataStreamSource);

        //设置水位线
        SingleOutputStreamOperator<JSONObject> waterStream = WaterStream(eltedStream);


        //3）按照唯一键去重

        KeyedStream<JSONObject, String> keyBy = waterStream.keyBy(x -> x.getString("id"));


        //4）转换数据结构  JSONObject转换为实体类TradeProvinceOrderWindow。 //累计下单次数 累计下单金额
        SingleOutputStreamOperator<TradeProvinceOrderBean> keySteam = KeySteam(keyBy);



        //6）按照省份ID分组  provinceId可以唯一标识数据。
        KeyedStream<TradeProvinceOrderBean, String> keyedStream = keySteam.keyBy(x -> x.getProvinceId());


        //7）开窗 聚合计算  度量字段求和，并在窗口闭合后补充窗口起始时间、结束时间以及当前统计日期。

        SingleOutputStreamOperator<TradeProvinceOrderBean> reduce = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(TradeProvinceOrderBean t1, TradeProvinceOrderBean t2) throws Exception {
                        t1.setOrderCount(t1.getOrderCount() + t2.getOrderCount());
                        t1.setOrderAmount(t1.getOrderAmount().add(t2.getOrderAmount()));

                        return t1;
                    }
                }, new WindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TradeProvinceOrderBean> iterable, Collector<TradeProvinceOrderBean> collector) throws Exception {

                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        Iterator<TradeProvinceOrderBean> iterator = iterable.iterator();
                        TradeProvinceOrderBean tradeProvinceOrderBean = iterator.next();

                        tradeProvinceOrderBean.setEdt(DateFormatUtil.tsToDateTime(end));
                        tradeProvinceOrderBean.setStt(DateFormatUtil.tsToDateTime(start));
                        tradeProvinceOrderBean.setCurDate(DateFormatUtil.tsToDateTime(System.currentTimeMillis()));

                        collector.collect(tradeProvinceOrderBean);
                    }
                });


//        reduce.print();


        //9）关联省份信息 补全省份名称字段。
        SingleOutputStreamOperator<TradeProvinceOrderBean> Mapstream = reduce.map(new RichMapFunction<TradeProvinceOrderBean, TradeProvinceOrderBean>() {

            private Connection connection;

            @Override
            public void open(Configuration parameters) throws Exception {
                connection = HBaseUtil.getHBaseConnection();
            }

            @Override
            public void close() throws Exception {
                connection.close();
            }

            @Override
            public TradeProvinceOrderBean map(TradeProvinceOrderBean tradeProvinceOrderBean) throws Exception {
                String provinceId = tradeProvinceOrderBean.getProvinceId();

                JSONObject dimBaseProvince = HBaseUtil.getCells(connection, Constant.HBASE_NAMESPACE, "dim_base_province", provinceId);

                String provincename = dimBaseProvince.getString("name");

                tradeProvinceOrderBean.setProvinceName(provincename);

                return tradeProvinceOrderBean;
            }
        });


        Mapstream.print();
        //10）写出到Doris。

        Mapstream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSinks(Constant.DWS_TRADE_PROVINCE_ORDER_WINDOW));

    }

    private static SingleOutputStreamOperator<TradeProvinceOrderBean> KeySteam(KeyedStream<JSONObject, String> keyBy) {
         return  keyBy.process(new ProcessFunction<JSONObject, TradeProvinceOrderBean>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, TradeProvinceOrderBean>.Context context, Collector<TradeProvinceOrderBean> collector) throws Exception {

                Long ts = jsonObject.getLong("ts");
                String provinceId = jsonObject.getString("province_id");

                String tsToDate = DateFormatUtil.tsToDateTime(ts);

                HashSet<String> set = new HashSet<>();
                set.add(jsonObject.getString("order_id"));

                collector.collect(TradeProvinceOrderBean.builder()
                        .curDate(tsToDate)
                        .ts(ts)
                        .provinceId(provinceId)
                        .orderCount(1L)
                        .orderAmount(jsonObject.getBigDecimal("split_total_amount"))
                        .orderIdSet(set)
                        .orderDetailId(jsonObject.getString("id"))
                        .build());
            }
        });

    }

    private static SingleOutputStreamOperator<JSONObject> WaterStream(SingleOutputStreamOperator<JSONObject> eltedStream) {
         return eltedStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getLong("ts")).withIdleness(Duration.ofSeconds(1)));
    }

    private static SingleOutputStreamOperator<JSONObject> eltStream(DataStreamSource<String> dataStreamSource) {
         return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);
                    if (jsonObject != null) {

                        Long ts = jsonObject.getLong("ts");
                        String provinceId = jsonObject.getString("province_id");
                        String orderId = jsonObject.getString("order_id");

                        if (ts != null && provinceId != null && orderId != null) {
                            jsonObject.put("ts", ts * 1000);

                            collector.collect(jsonObject);
                        }
                    }
                }
            }
        });
    }
}
